#include "raft.h"
#include <assert.h>
#include <random>
#include <sstream>
#include <algorithm>
#include <string>


#define PRINT(...) PrintOutput(m_factory, m_id, __func__, ##__VA_ARGS__)
template <typename T>
void osspack(std::ostream &o, T &&t) { o << t; }
template <typename T, typename... Args>
void osspack(std::ostream &o, T &&t, Args &&...args)
{
    o << t;
    osspack(o, std::forward<Args>(args)...);
}
template <typename... Args>
void PrintOutput(std::shared_ptr<raft::objfactory<raft::server>> factory, int id, const std::string &func, Args &&...args)
{
    auto server = factory->Get(id, factory);
    // 将不同类型的数据格式化为一个字符串
    std::ostringstream o;
    std::unordered_map<int, char> mp;
    mp[0]='N';
    mp[1]='L';
    mp[2]='C';
    mp[3]='F';
    osspack(o, " (id:", id, " term: ", server->Term(), " state: ", mp[(int)server->GetState()], ")  ", func, "->", std::forward<Args>(args)...);
    factory->Get(0, factory)->AddPrint(o.str());
}


// follower
int raft::server::followerpKeepAlive()
{
    std::cout << getTime() << "CONNECTING..." << std::endl;
    _net.close();

    // try to connect with the coordinator
    while (_net.reaccept())
    {
        // sleep(1);
     }

     std::cout << getTime() << "CONNECTED..." << std::endl;


    return KV_OK;
}

int raft::server::followerWorking()
{   
    AppendEntriesReply reply1;
    
    VoteArgs args2;
    std::cout << getTime() << "WORKING..." << std::endl;
    int rc = KV_OK;
    std::string type;
    std::string content;
    std::string rmsg;
    std::vector<std::string> comds;

    // if network close will return error
    rc = _net.recv(type);
    if(DE_BUG) 
        std::cout <<"rc:"<< rc << "\trecv: " << type.c_str() << std::endl;
    if(type == "AppendEntriesArgs") {
        AppendEntriesReply reply;
        AppendEntriesArgs args;
        rc = _net.recv(content);
        if(DE_BUG) 
            std::cout <<"rc:"<< rc << "\trecv: " << content.c_str() << std::endl;
        
        std::string sendContent = "";
        args = deserializeForAEA(content);
        // std::cout << "recv args success " << std::endl;
        reply = RecvAppendEntriesAndReply(args);
        // std::cout << "reply success " << std::endl;
        sendContent = serializeForAER(reply);
        _net.send(sendContent);
    }
    else if(type == "VoteArgs") {
        VoteReply reply;
        VoteArgs args;
        rc = _net.recv(content);
        if(DE_BUG) 
            std::cout <<"rc:"<< rc << "\trecv: " << content.c_str() << std::endl;
        std::string sendContent = "";
        args = deserializeForVA(content);
        std::cout << "recv args success " << std::endl;
        reply = RequestVoteAndReply(args);
        std::cout << "reply success " << std::endl;
        sendContent = serializeForVR(reply);
        _net.send(sendContent);
    }
    return rc;
}


std::string raft::server::eventParser(std::string event)
{
    std::string rstr = _parser.getErrorMessage();
    int delnumber = 0;
    
    std::vector<std::string> comds;
    // parse and send
    _parser.parserRESPArry((char *)event.c_str(), comds);
    if(comds.size() < 1) {
        rstr = _parser.getErrorMessage();
        goto done;
    }

    switch(comds[0][0]) {
        case 'D': // DEL CS06142 CS162
            comds.erase(comds.begin());
            delnumber = _kvdb.kvDELArry(comds);
            rstr = _parser.getIntergerMessage(delnumber);
            break;
            
        case 'S': // SET key value
            if(comds.size() != 3) {
                rstr = _parser.getErrorMessage();
            } else {
                _kvdb.kvSET(comds[1], comds[2]);
                rstr = _parser.getSuccessMessage();
            }
            break;
        case 'G': // GET key
            if(comds.size() != 2) {
                rstr = _parser.getErrorMessage();
            } else {
                rstr = _parser.getRESPArry( _kvdb.kvGET(comds[1]) );
            }
            break;
        case 'P': // P
            rstr = "PRE";
            break;
        default:
            rstr = _parser.getErrorMessage();
            break;

    }

done:
    return rstr;
}
void raft::server::LogToKVStore() {
    int logID = 0;
    while(1) {
        std::this_thread::sleep_for(std::chrono::milliseconds(1000)); //300
        // std::cout << "follower id: " << m_id <<" log Update! " << std::endl;
        // std::cout << "logID: " << logID << " log size: " << m_log_vec.size() << " m_commit_index: " << m_commit_index << std::endl;
        // std::unique_lock<std::mutex> _(m_mutex);
        // _raftMutex.get();
        if(m_log_vec.empty())
            continue;
        if(logID >= (int)m_log_vec.size())
            continue;
        if(DE_BUG)
            std::cout << "id: " << m_id <<" log Update! " << std::endl;
        if(DE_BUG)
            std::cout << "logID: " << logID << " log size: " << m_log_vec.size() << " m_commit_index: " << m_commit_index << std::endl;
        Log l = m_log_vec[logID];
        std::string event = l.event;
        if(DE_BUG) 
            std::cout << "event: " << event << std::endl;
        if(event.find("\r\n") == std::string::npos) {
            logID++;
            if(DE_BUG) {
                std::cout << "election event, ignore!" << std::endl;
            }
            continue;
        }
        else {
            if(DE_BUG) {
                std::cout << "KVStore event, parser!" << std::endl;
            }
            if(eventParser(event) != _parser.getErrorMessage()) {
                logID++;
                if(DE_BUG){
                    std::cout << "eventParser success" << std::endl;
                    std::cout << "follower id: " << m_id <<" KVStore Update! " << std::endl;
                    KVPrint();
                }
            }
            else {
                if(DE_BUG) {
                    std::cout << "eventParser error\n" << std::endl;
                }
            }
        }
        // _raftMutex.release();
    }
}

void raft::server::LeaderpKeepAlive(int id)
{
    assert(_pnet.count(id) == 1);
    while (1)
    {
        if (_pstate[id] != P_WAITING)
        {
            continue;
        }
        if(DE_BUG)
            std::cout << getTime() << _pinfo[id].add << ":" << _pinfo[id].port <<" CONNECTING..." << std::endl;
        // try to connect with the participant
        while (_pnet[id].reconnect())
        {
            if(DE_BUG)
                std::cout << "reconnecting..." << std::endl;
            // sleep(1);
            // _pstate[id] = P_WAITING;
        }
        if(DE_BUG)
            std::cout << getTime() << _pinfo[id].add << ":" << _pinfo[id].port <<" CONNECTED" << std::endl;
        _pstate[id] = P_TEMP;
    }

}
std::string raft::server::Request(std::string input) {
    std::string key = input;
    std::string value = _kvdb.kvGET(key);
    // int res = std::stoi(value);
    return value;
}
int raft::server::Working()
{
    std::cout << getTime() << "WORKING FOR CLIENT..." << std::endl;
    std::vector<std::string> task;
    while (_net.acceptWithoutCloseBind())
    {
        if(DE_BUG) 
            std::cout << "accept without close bind..." << std::endl;
        _net.initBind();
    }
    _net.recvCommand(task);
    if(DE_BUG){
        printf("TASK:\n");
        for(auto i : task)
        printf("%s\n", i.c_str());
    }

    std::string resp = _parser.getRESPArry(task);

    std::string rc = _parser.getErrorMessage();
    if(DE_BUG){
        printf("RESP:\n");
        printf("%s\n", resp.c_str());
    }
    // _workingMutex.get();
    rc = AddLog(resp);
    
    if (task[0] == "SET" || task[0] == "DEL")
    {
        rc = "+OK";
        // rc = UpdateDB(resp);
    }
    else if (task[0] == "GET")
    {
        rc  = Request(task[1]);
    }
    else
    {
        rc = _parser.getErrorMessage();
    }
    rc = _parser.getRESPArry(rc);
    std::cout << "task success, rc: " << rc << std::endl;
    // _workingMutex.release();
    
    _net.sendResult(rc);
    sleep(0.1);
    _net.close();

    return KV_OK;
}
void raft::server::work()
{
    while(1) 
    {
        if(m_state == State::Leader)
            Working();
    }
}

// other

void raft::server::KVPrint() {
    std::cout << "follower id: " << m_id <<" KVStore Print:" << std::endl;
    _kvdb.kvPrint();
}

// init leadet when first start
void raft::server::Init(std::vector<NodeInfo> ps, NodeInfo c)
{
    _TXID   = TXID_START;
    // _cstate = C_RECOVERY;
    _net    = Network(c.port);
    
    _fnum = ps.size(); 

    for (int id = 1; id <= (int)_fnum; id++)
    {
        // init p's state is waiting
        _pstate[id] = P_State::P_WAITING;
        // set the p's network
        // if(id == m_id) continue;
        _pnet[id] = Network(ps[id - 1].add, ps[id - 1].port);
        // set the node info
        _pinfo[id] = ps[id - 1];
        if(DE_BUG) {
            if(_pnet.count(id) != 1) {
                std::cout << "pnet id: " << id << " port: " << ps[id - 1].port << " set failed" << std::endl;
            }
            else  {
                std::cout << "pnet id: " << id << " port: " << ps[id - 1].port << std::endl;
            }
        }
    }
    // init leader

    // if(c.port == 8001) {
    //     std::cout << c.port << " I am the leader!" << std::endl;
    //     ToLeader();
    // }
    // else {
    //     ToFollower(0, 1);
    // }
    // ToFollower(0, 1);
    auto print = m_factory->Get(0, m_factory);
    tpool.submit([print]
                 { print->Print(); });

    // std::cout << "m_next_index_vec.size() " << m_next_index_vec.size() << std::endl;

}

int raft::server::Launch()
{
    State state = m_state;
    while (true)
    {
    std::this_thread::sleep_for(std::chrono::milliseconds(300));
    // std::unique_lock<std::mutex> _(m_mutex);
    // if(state == m_state) 
    //     continue;
    // _raftMutex.get();
    if(m_state == State::None) {
        PRINT("State is None, can not launch");
        return KV_ERR;
    }
    if(m_state == State::Leader) {
        if(G_DE_BUG) {
            PRINT("Launch as Leader");
        }
        state = State::Leader;
        // std::cout << _fnum << std::endl;
        for(int id = 1; id <= _fnum; id++) {
            if(id == m_id) continue;
            if(DE_BUG)
                std::cout << "leader keepalive follower id: " << id << std::endl;
            _talive[id] = std::thread(&raft::server::LeaderpKeepAlive, this, id);
        }
        // if(DE_BUG)
        //     std::cout << "Leader start work with client" << std::endl;
        // _workThread = std::thread(&raft::server::work, this);

        // waiting all the p start for a while
        sleep(2);
        

        for(int id = 1; id <= _fnum; id++) {
            if(_talive[id].joinable())
                _talive[id].join();
        }
        // _workThread.join();
    }
    else {
        if(G_DE_BUG)
            PRINT("Launch as Follower");
        state = State::Follower;
        // followerpKeepAlive();
        // _fkeep = std::thread(&raft::server::followerpKeepAlive, this);
        // _fwork = std::thread(&raft::server::followerWorking, this);
        // sleep(1);

        // if(_fkeep.joinable())
        //     _fkeep.join();
        // if(_fwork.joinable())
        //     _fwork.join();
        followerpKeepAlive();
        while(1) {
            if(DE_BUG)
                std::cout << "follower working " << m_id <<std::endl;
            if(followerWorking()) {
                followerpKeepAlive();
            }
        }
        
    }
    // _raftMutex.release();
    }
    return KV_OK;
}